package buaa.qel.rpc.client.core;

import buaa.qel.rpc.common.info.RegistryNodeInfo;
import buaa.qel.rpc.common.zookeeper.CuratorClient;
import buaa.qel.rpc.common.zookeeper.config.ZookeeperConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * 服务发现：在监听并获取zookeeper节点上注册的服务后更新connectionManager的注册信息集合
 */
@Slf4j
public class ServiceDiscovery {
    private CuratorClient curatorClient;

    public ServiceDiscovery(String registryAddress) {
        this.curatorClient = new CuratorClient(registryAddress);
        discoverService();
    }

    private void discoverService() {
        try {
            log.info("Start to connect available services...");
            getServiceAndUpdateServer();
            log.info("Connected and watch child nodes");
            curatorClient.watchPathChildrenNode(ZookeeperConstants.ZK_REGISTRY_PATH,
                (CuratorFramework curatorFramework, PathChildrenCacheEvent evt) -> {
                    PathChildrenCacheEvent.Type type = evt.getType();
                    ChildData childData = evt.getData();
                    switch (type) {
                        case CONNECTION_RECONNECTED:
                            log.info("Reconnected to zk, try to get latest service list");
                            getServiceAndUpdateServer();
                            break;
                        case CHILD_ADDED:
                            getServiceAndUpdateServer(childData, PathChildrenCacheEvent.Type.CHILD_ADDED);
                            break;
                        case CHILD_UPDATED:
                            getServiceAndUpdateServer(childData, PathChildrenCacheEvent.Type.CHILD_UPDATED);
                            break;
                        case CHILD_REMOVED:
                            getServiceAndUpdateServer(childData, PathChildrenCacheEvent.Type.CHILD_REMOVED);
                            break;
                    }
                }
            );
        } catch (Exception ex) {
            log.error("Watch node exception: " + ex.getMessage());
        }
    }

    /**
     * 获取zookeeper的所有子节点路径，调用connectionManager
     */
    private void getServiceAndUpdateServer() {
        try {
            List<String> nodeList = curatorClient.getChildren(ZookeeperConstants.ZK_REGISTRY_PATH);
            List<RegistryNodeInfo> dataList = new ArrayList<>();
            for (String node : nodeList) {
                log.debug("Service node: " + node);
                byte[] bytes = curatorClient.getData(ZookeeperConstants.ZK_REGISTRY_PATH + "/" + node);
                String json = new String(bytes);
                RegistryNodeInfo registryNodeInfo = RegistryNodeInfo.fromJson(json);
                dataList.add(registryNodeInfo);
            }
            log.debug("Service node data: {}", dataList);
            UpdateConnectedServer(dataList);
        } catch (Exception e) {
            log.error("Get node exception: " + e.getMessage());
        }
    }

    private void getServiceAndUpdateServer(ChildData childData, PathChildrenCacheEvent.Type type) {
        String path = childData.getPath();
        String data = new String(childData.getData(), StandardCharsets.UTF_8);
        log.info("Child data updated, path:{},type:{},data:{},", path, type, data);
        RegistryNodeInfo registryNodeInfo =  RegistryNodeInfo.fromJson(data);
        updateConnectedServer(registryNodeInfo, type);
    }

    private void UpdateConnectedServer(List<RegistryNodeInfo> dataList) {
        ConnectionManager.getInstance().updateConnectedServer(dataList);
    }


    private void updateConnectedServer(RegistryNodeInfo registryNodeInfo, PathChildrenCacheEvent.Type type) {
        ConnectionManager.getInstance().updateConnectedServer(registryNodeInfo, type);
    }

    public void stop() {
        this.curatorClient.close();
    }
}
